package com.imooc.flink.mysql;

import com.imooc.flink.util.MySQLUtils;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;

/**
 * @author: Chen Yixing
 * @since: 2023/10/16 14:55:18
 * @description:
 */
public class MysqlSource extends RichSourceFunction<Tuple2<Long, String>> {
    Connection connection;
    PreparedStatement selectPstmt;

    @Override
    public void open(Configuration parameters) throws Exception {
        connection = MySQLUtils.getConnection();
        selectPstmt = connection.prepareStatement("select * from  t_data_center_test");
    }

    @Override
    public void run(SourceContext<Tuple2<Long, String>> sourceContext) throws Exception {
        ResultSet resultSet = selectPstmt.executeQuery();
        while (resultSet.next()) {
            long id = resultSet.getLong("id");
            String value = resultSet.getString("name");
            Tuple2<Long, String> dbResult = Tuple2.of(id, value);
            sourceContext.collect(dbResult);
        }
    }

    @Override
    public void cancel() {

    }

    @Override
    public void close() throws Exception {
        if (selectPstmt != null) {
            selectPstmt.close();
        }
        if (connection != null) {
            connection.close();
        }
    }
}
